package ubc.worker;

import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import ubc.*;


/* Client:
 * - If we run out of uncrawled nodes, ask the server for more and send to the server our crawled nodes
 * - Crawl uncrawled nodes
 * - Fully implements non-blocking io
 */

public class Client extends Thread{

    int _timeout; 
    int _numConnections = 20;
    String serverHostPort;
    int _mode = ubc.Constants.FULL_MODE;

    QueueOrganizer qo;
    ByteBuffer serverResponseBuf = ByteBuffer.allocate(10000);

    Selector selector;
    InetSocketAddress _isa;

    List<CrawlResult> tempResults; // Keep the results temporaily, make sure we were able to send the results to the server before we delete them

    Boolean triedReading = false; // Used to check whether we have tried reading yet

    long lastSentTime = 0;

    // Constructor
    public Client(String hostport, int timeout, int numConnections) throws Exception {

        _timeout = timeout;
        _numConnections = numConnections;
        serverHostPort = hostport;

        qo = new QueueOrganizer("Client", ubc.Constants.clientFilePath);
        qo.start();
        
        selector = Selector.open();
        connect(serverHostPort, ByteBuffer.allocate(1));

        // Quick bootstrap node for debugging
//        qo.uncrawledHosts.enqueue("74.108.0.32:23589"); // Has Files
//        qo.uncrawledHosts.enqueue("reala.ece.ubc.ca:5627");
//        qo.uncrawledHosts.enqueue("208.66.72.202:9228"); // No Files
    }

    public void connect(String hostport, Object attachment) throws Exception{

        String host = hostport.split(":")[0];
        int port = Integer.parseInt(hostport.split(":")[1]);

        InetAddress ia = InetAddress.getByName(host);
        InetSocketAddress isa = new InetSocketAddress(ia, port);
        
        System.out.println("Connecting to: " + isa.getAddress() + ":" + isa.getPort() + "...");
        SocketChannel client = SocketChannel.open();
        client.configureBlocking(false);
        client.connect(isa);
        client.register(selector, SelectionKey.OP_CONNECT, attachment );
    }

    public void run () {

        while(!Thread.interrupted()) {

            // Waits half a sec for an event
            try {
                selector.select(500);
            } catch (Exception e) {
                System.err.println("Client.select().Exception");
                //e.printStackTrace();
            }
            Set keys = selector.selectedKeys();
            Iterator i = keys.iterator();
            while(i.hasNext() && !Thread.interrupted()) {
                SelectionKey key = (SelectionKey)i.next();
                i.remove();

                if(key.isConnectable()) {
                    finishConnect(key);
                }
                if( !key.isValid() ) {

                    continue;
                }
                if (key.isWritable()) {
                    if( ByteBuffer.class.isInstance(key.attachment()) ) {// Only a Connection with the server contains a ByteBuffer attachment
                        long curTime = System.currentTimeMillis();
                        if( curTime-lastSentTime > 2000+(new java.util.Random()).nextInt(6000) ) {
                            writeToServer(key);
                        }
                    }
                    else
                        writeToGnutella(key);
                }
                else if (key.isReadable()) {
                    if( ByteBuffer.class.isInstance(key.attachment()) ) // Only a Connection with the server contains a ByteBuffer attachment
                        readFromServer(key);
                    else
                        readFromGnutella(key);
                }
            }


            // Check if we have created a connection to the server, if we don't, then create one
            // Note, just because we have a connection does not mean we are connected yet
            SelectionKey k = getServerKey(selector); // Get the key thats associated with the server
            if(k == null && // No Connection to the server
               qo.uncrawledHosts.size() == 0 && // No uncrawled hostnames left
               triedReading == true) { // We have tried reading (w/o this line, the client will constantly write and will have no chance to read)
                try {
                    connect(serverHostPort, ByteBuffer.allocate(1));
                } catch(Exception e) {
                    System.out.println("run.serverConnection.Exception");
                    //e.printStackTrace();
                }
            }

            // Connect to new nodes if our total number of current connections is less than _numConnections
            while ( (selector.keys().size()<_numConnections+1) && (qo.uncrawledHosts.size()>0) ) {
                try {
                    String addr = (String)qo.uncrawledHosts.dequeue();
                    connect( addr, new CrawlResult(addr) );
                } catch (Exception e) {
                    System.out.println("run.clientConnection.Exception");
                    //e.printStackTrace();
                }
            }

            // Check if any nodes has timed out
            long curTime = System.currentTimeMillis();
            for(int j=0; j<selector.keys().size(); j++) {
                Set<SelectionKey> sk = selector.keys();
                Iterator iter = sk.iterator();
                while(iter.hasNext()) {
                    SelectionKey sKey = (SelectionKey)iter.next();

                    if( CrawlResult.class.isInstance(sKey.attachment()) ) {
                        long startTime = ((CrawlResult)sKey.attachment()).getStartTime();
                        if( (curTime - startTime) > _timeout*1000 ) {
                            CrawlResult cr = (CrawlResult)sKey.attachment();
                            cr.setStatus(CrawlResult.TIMEOUT_EXCEPTION);
                            int stage = cr.getStage();
                            // If we are in the process of getting files list, then make sure we don't send a corrupted version
                            if(stage == CrawlResult.SENT_FILES_REQUEUST)
                                cr.clearFilesList();
                            qo.crawledResults.enqueue(cr);
                            ubc.nio.close(sKey);
                        }
                    }
                }
            }


        }
    }

    // returns the key that has the given attachment
    public SelectionKey getServerKey(Selector selector) {
        int numKeys = selector.keys().size();
        for(int i=0; i<numKeys; i++) {
            Set<SelectionKey> sk = selector.keys();
            Iterator iter = sk.iterator();
            while(iter.hasNext()) {
                SelectionKey sKey = (SelectionKey)iter.next();
                Object obj = sKey.attachment();
                if( ByteBuffer.class.isInstance(obj) ) {
                    return sKey;
                }
            }
        }
        return null;
    }

    // Returns true if we are connected to the server
    public Boolean finishConnect(SelectionKey key) {
        SocketChannel channel = (SocketChannel) key.channel();
        try {
            if( channel.finishConnect() ) {
                key.interestOps(SelectionKey.OP_WRITE);
                return true;
            }
        } catch (ConnectException e) {
//            System.out.println("finishConnect.ConnectException");
            //e.printStackTrace();
            if ( CrawlResult.class.isInstance(key.attachment())) {
                CrawlResult cr = (CrawlResult)key.attachment();
                cr.setStatus(CrawlResult.CONNECT_EXCEPTION);
                qo.crawledResults.enqueue(cr);
                //e.printStackTrace();
            }
            key.cancel();
            return false;
        } catch (Exception e) {
//            System.err.println("finishConnect.Exception");
            //e.printStackTrace();
            key.cancel();
            return false;
        }
        return false;
    }

    // Sends some crawled CrawlResults to the server
    public void writeToServer(SelectionKey key) {

        // Cleanly finish and stop writing to the file
        qo.stopWriter();

        // Read the file that contains our list of crawled CrawlResults
        ObjectReader or = new ObjectReader(qo._path);
        tempResults = (List<CrawlResult>)or.readAll();
        CrawlWrapper wrappedObj = new CrawlWrapper(tempResults, _mode);
        try {
            // Send the results
            int numWrite = ubc.nio.writeIncludingLength(key, wrappedObj);
            wrappedObj = null;
            lastSentTime = System.currentTimeMillis();

            // If we lost the connection to the server, then add the results back to our queue
            if(numWrite == -1) {
                Iterator i = tempResults.iterator();
                while(i.hasNext()) {
                    qo.crawledResults.enqueue(i.next());
                }
                ubc.nio.close(key);
                return;
            }
            
            // Print out some info regarding what was sent
            System.err.println("Num Results Sent: " + tempResults.size());
//            System.err.print("Sent Result For: [");
//            for(int i=0; i<tempResults.size(); i++) {
//                System.err.print(tempResults.get(0).getHostPort() + ", ");
//            }
//            System.err.print("] \n");

            // Set our interest to read
            key.interestOps(SelectionKey.OP_READ);
        }

        // If we failed to send, store the results back to the queue
        catch (Exception e) {
            wrappedObj = null;
            System.out.println("writeToServer.Exception");
            //e.printStackTrace();
            ubc.nio.close(key);
            Iterator i = tempResults.iterator();
            while(i.hasNext()) {
                qo.crawledResults.enqueue(i.next());
            }
        }

        // Start writing to the file again
        qo.startWriter();

    }

    // Sends a request message to the gnutella node associated with the key
    public void writeToGnutella(SelectionKey key) {

        CrawlResult cr = (CrawlResult)key.attachment();

        // We have not started crawling yet, lets send a peers-request message
        if( cr.getStage() == CrawlResult.NOT_CRAWLED ) {
            String request_nodes  = "GNUTELLA CONNECT/0.6\r\n" +
                        "User-Agent: UBC_CRAWLER (crawl)\r\n" +
                        "Query-Routing: 0.2\r\n"+
                        "X-Ultrapeer: False\r\n"+
                        "Crawler: 0.1\r\n" +
                        "\r\n";
            try {
                ubc.nio.writeString(key, request_nodes);
            } catch(Exception e) {
                System.out.println("writeToGnutella.NOT_CRAWLED.Exception");
                //e.printStackTrace();
                cr.setStatus(CrawlResult.SEND_EXCEPTION);
                qo.crawledHosts.enqueue(cr);
                ubc.nio.close(key);
            }
//            System.out.println("Sent PEERS request to " + ((SocketChannel)key.channel()).socket().getInetAddress() );
            cr.setStage(CrawlResult.SENT_PEERS_REQUEUST);
            key.attach(cr); // Attach the newly updated CrawlResult back to the key
            key.interestOps(SelectionKey.OP_READ); // Set our interest to read
        }

        // We have already requested and received the list of peers,
        // lets send an Ack to close the connection
        else
        if( cr.getStage() == CrawlResult.GOT_PEERS_RESPONSE ) {
            String ack = "GNUTELLA/0.6 200 OK\r\n \r\n";
            try {
                ubc.nio.writeString(key, ack);

                // If we are in Full Mode, make a new connection to the node to crawl the files
                if( _mode == CrawlWrapper.FULL_MODE ) {

                    cr.buf.clear(); // Clear the buffer for the next stage
                    cr.setStage(CrawlResult.ACKED_PEERS_RESPONSE);

                    // Get the node's host and port then close the connection
                    String addr = ((SocketChannel)key.channel()).socket().getInetAddress().getHostAddress();
                    int port = ((SocketChannel)key.channel()).socket().getPort();
                    ubc.nio.close(key);

                    // Make a new connection to the node to crawl the files
                    try {
                        connect(addr+":"+port, cr); // include the current CrawlResult as the attachment
                    } catch (Exception e) {
                        System.out.println("writeToGnutella.GOT_PEERS_RESPONSE.Exception 1");
                        //e.printStackTrace();
                        cr.setStatus(CrawlResult.CONNECT_EXCEPTION);
                        ubc.nio.close(key);
                    }
                } else { // We are in Minimal Mode, add the result to our queue and close the connection
                    cr.setStatus(CrawlResult.CRAWLED);
                    qo.crawledResults.enqueue( cr );
                    ubc.nio.close(key);
                }

            } catch(Exception e) {
                System.out.println("writeToGnutella.GOT_PEERS_RESPONSE.Exception 2");
                cr.setStatus(CrawlResult.SEND_EXCEPTION);
                qo.crawledHosts.enqueue(cr);
                ubc.nio.close(key);
            }

        }

        // We have already requested and received the list of peers, and sent an ack,
        // lets request for the files list
        else
        if( cr.getStage() == CrawlResult.ACKED_PEERS_RESPONSE ) {
            String localAddress=((SocketChannel)key.channel()).socket().getLocalAddress().getHostAddress();
            String request_content = "GET / HTTP/1.1\r\n" +
                        "Host: "+localAddress+"\r\n"+
                        "User-Agent: UBC_CRAWLER/0.1\r\n" +
                        "Accept: text/html, application/x-gnutella-packets\r\n"+
                        "Connection: close\r\n"+
                        "\r\n";
            try {
                ubc.nio.writeString(key, request_content);
            } catch(Exception e) {
                System.out.println("writeToGnutella.ACKED_PEERS_RESPONSE.Exception");
                //e.printStackTrace();
                cr.setStatus(CrawlResult.SEND_EXCEPTION);
                qo.crawledHosts.enqueue(cr);
                ubc.nio.close(key);
            }
//            System.out.println("Sent FILES request to " + ((SocketChannel)key.channel()).socket().getInetAddress() );
            cr.setStage(CrawlResult.SENT_FILES_REQUEUST);
            key.attach(cr); // Attach the newly updated CrawlResult back to the key
            key.interestOps(SelectionKey.OP_READ); // Set our interest to read
        }

    }

    // Reads from the server channel, if the entire message has been read,
    // it will add the hostnames from the message into our queue of uncrawledHosts
    public void readFromServer(SelectionKey key) {
        ByteBuffer buf = (ByteBuffer)key.attachment();
        try {
            // If we haven't gotten the size of the entire data yet, then find out the size
            if ( buf.capacity() == 1 ) {
                buf = ByteBuffer.allocate(4);
                int totalRead = 0;
                long startTime = System.currentTimeMillis();
                while(true) {
                    long curTime = System.currentTimeMillis();
                    if( (curTime-startTime) > 5000)
                        return;
                    int numRead =  ubc.nio.read(key, buf);
                    if(numRead == -1) {
                        Iterator i = tempResults.iterator();
                        while(i.hasNext()) {
                            qo.crawledResults.enqueue(i.next());
                        }
                        return;
                    }
                    totalRead += numRead;
                    if(totalRead == 4)
                        break;
                }
//                tempResults=null;
                buf.clear();
                int msgLength = buf.getInt();
                buf = ByteBuffer.allocate(msgLength);
                buf.clear();
            }

            // Read some data
            int numRead = ubc.nio.read(key, buf);

            // Attach the newly updated buf back to the key
            key.attach(buf);

            // If we have read the entire data, then process it
            if( buf.position() == buf.capacity() ) {
                handleServerResponse(buf);
                buf = ByteBuffer.allocate(1);
                key.attach(buf);
                ubc.nio.close(key);
            }
        } catch (Exception e) {
            System.out.println("readFromServer.Exception");
            e.printStackTrace();
            ubc.nio.close(key);
        }
    }

    // Once the buf has the entire data, parse the buf and add the hostnames to our queue
    public void handleServerResponse(ByteBuffer buf) {
        triedReading = true;
        CrawlWrapper cw = (CrawlWrapper)ubc.nio.toObject(buf.array());

        // If we received a quit message, then quit
        if( cw.getMessage() == CrawlWrapper.QUIT) {
            System.out.println("Server told me to QUIT");
            this.interrupt();
        } else { // Otherwise, set our mode to the mode of the server
            _mode = cw.getMessage();
        }
        
        // Add the newly received hostnames to our list of uncrawledHosts
        Queue q = (Queue)cw.getObject();
        
        if (q!=null) {
//            System.err.println("Received Hostnames: " + q.toString());
            System.err.println("Num Host Received: " + q.size());
            if (q.size() == 0 && selector.keys().size() == 0) {
                try {
                    System.out.println("Sleeping...");
                    Thread.sleep(2000+(new java.util.Random()).nextInt(6000));
                } catch (Exception e) {

                }
            }
            qo.uncrawledHosts.set(q);
        } else {
            System.err.println("Num Host Received: 0");
            try {
                this.sleep(1000+(new java.util.Random()).nextInt(4000));
                System.out.println("slept");
            } catch (Exception e) {

            }
//            System.err.println("Received Hostnames: []");
        }
        
        
    }

    // Reads from the gnutella node channel, if we receive a EOF (-1), then
    // we can parse the result.
    public void readFromGnutella(SelectionKey key) {

        CrawlResult cr = (CrawlResult)key.attachment();
        try {
            if( cr.getStage() == CrawlResult.SENT_PEERS_REQUEUST ) {
                
                int numRead = ubc.nio.read(key, cr.buf);
                key.attach(cr); // Attach the newly updated buf back to our key
                if( numRead == -1 ) { // EOF
                    handleGnutellaPeersResponse(key);
                }
            }
            if( cr.getStage() == CrawlResult.SENT_FILES_REQUEUST ) {
                
                int numRead = ubc.nio.read(key, cr.buf);
                key.attach(cr); // Attach the newly updated buf back to our key
                if( numRead == -1 ) { // EOF
                    handleGnutellaContentResponse(key);
                }
            }
            
        } catch(Exception e) {
            System.out.println("readFromGnutella.Exception");
            //e.printStackTrace();
            cr.setStatus(CrawlResult.IO_EXCEPTION);
            qo.crawledResults.enqueue(cr);
            ubc.nio.close(key);
        }
    }

    // Once the buf has the entire data, parse the buf and add the ultrapeers and leaves
    // If we are in full mode, make a new connection to the node to crawl the files
    // If we are in minimal mode, then add the result to our queue and close the connection
    public void handleGnutellaPeersResponse(SelectionKey key) {

        // Parse the buf
        CrawlResult cr = (CrawlResult)key.attachment();
        String response = new String(cr.buf.array());
        if( !response.isEmpty() ) {
            parsePeers(response, cr );
        }
        cr.setStage(CrawlResult.GOT_PEERS_RESPONSE);
        key.attach(cr);
        key.interestOps(SelectionKey.OP_WRITE);
    }
    

    // Once the buf has the entire data, parse the buf for the file sizes and names, 
    // then add the result to our queue and close the connection
    public void handleGnutellaContentResponse(SelectionKey key) {
        CrawlResult cr = (CrawlResult)key.attachment();
        int headerIndex = findEndOfHeader(cr.buf.array());
        parseFilesListHeader(cr.buf.array(), headerIndex, cr);
        int curIndex = headerIndex;
        while( true ) {
            byte[] qhitSizeBuf = subBuffer(cr.buf.array(),curIndex+19,curIndex+23);
            if ( qhitSizeBuf == null )
                break;
            int qhitSize = ByteOrder.leb2int(subBuffer(cr.buf.array(),curIndex+19,curIndex+23),0);
            if (qhitSize > 0 ) {
                parseQueryHit(subBuffer(cr.buf.array(), curIndex+23, curIndex+23+qhitSize), cr);
                curIndex += qhitSize + 23;
            } else
                break;
        }

        // Bruteforce method to get rid of a bug where a node with no files returns
        // a bunch of junk corresponding to ultrapeers and leaves
        // due to the inability to terminate the peer-request
        if( cr.getFilesList().size() > 10 &&
            cr.getFilesList().get( cr.getFilesList().size() -1 ).isEmpty() &&
            cr.getFilesList().get( cr.getFilesList().size() -2 ).isEmpty() )
            cr.clearFilesList();
        
        cr.setStatus(CrawlResult.CRAWLED);
        qo.crawledResults.enqueue(cr);
        ubc.nio.close(key);
    }

    // Prases the ultrapeer and leave names from response into the cResult
    public void parsePeers(String response, CrawlResult cResult){
        int beginIndex;
        int endIndex;
        String upeers=new String();
        String leaves=new String();

        beginIndex=response.indexOf("Peers:");
        endIndex=response.indexOf("\n",beginIndex);
        upeers=response.substring(beginIndex+6,endIndex);
        upeers=upeers.trim();

        beginIndex=response.indexOf("Leaves:");
        endIndex=response.indexOf("\n",beginIndex);
        leaves=response.substring(beginIndex+7,endIndex);
        leaves=leaves.trim();

        cResult.setUltrapeers(upeers);
        cResult.setLeaves(leaves);

        int strBegin=response.indexOf("User-Agent:");
        int strEnd=response.indexOf('\n',strBegin);
        String agent=response.substring(strBegin+12,strEnd);
        agent.trim();

        strBegin=agent.indexOf('/');
        if (strBegin==-1)
            strBegin=agent.indexOf(' ');

        agent=agent.substring(0,strBegin);
        cResult.setAgent(agent);
    }

    public static byte[] subBuffer(byte[] buffer,int beginIndex,int endIndex){

        if (beginIndex>buffer.length){
            return null;
        }
        if (endIndex>buffer.length){
            endIndex=buffer.length;
        }
        int bLen=endIndex-beginIndex;
        byte[] result=new byte[bLen];

        System.arraycopy(buffer,beginIndex,result,0,bLen);

        return result;
    }

    // Find the end of the header of the Gnutella Content response
    public static int findEndOfHeader(byte[] packet){

        for (int i=0;i<=packet.length-4;i++){
            if (packet[i]=='\r' && packet[i+1]=='\n' && packet[i+2]=='\r' && packet[i+3]=='\n'){
                return i+4;
            }
        }
        return -1;
    }

    // Parses the Gnutella Content response for the node's agent
    public void parseFilesListHeader(byte[] flist, int headerEndIndex, CrawlResult cResult){

        String header=new String(subBuffer(flist,0,headerEndIndex));
        int strBegin=header.indexOf("Server:");
        int strEnd=header.indexOf('\n',strBegin);
        header=header.substring(strBegin+8,strEnd);
        header.trim();

        strBegin=header.indexOf('/');
        if (strBegin==-1)
            strBegin=header.indexOf(' ');

        header=header.substring(0,strBegin);
        cResult.setAgent(header);

    }

    // Parses the bytes of a query-hit for the file names&sizes and adds them to the cResult
    public void parseQueryHit(byte[] qhit, CrawlResult cResult){
        int beginIndex;
        int endIndex;
        int numOfFiles=qhit[0];
        String sentList=new String();
        String filesList=new String();
        String name=new String();
        int size = 0;

        qhit=subBuffer(qhit,11,qhit.length);
        name=new String(qhit);

        byte[] fsize = new byte[4];
        byte[] fname=new byte[512];
        int num=0;
        while (qhit.length>0 && num<numOfFiles){

            // File Size
            qhit=subBuffer(qhit,4,qhit.length);
            endIndex=4;
            fsize=subBuffer(qhit,0,endIndex);
            size = ByteOrder.leb2int(fsize, 0);
            cResult.addFileSize(size);

            // File Name
            qhit=subBuffer(qhit,4,qhit.length);
            endIndex=bufferIndexOf(qhit,'\0');
            fname=subBuffer(qhit,0,endIndex);
            name=new String(fname);
            cResult.addFileName(name);
            qhit=subBuffer(qhit,endIndex+1,qhit.length);

            endIndex=bufferIndexOf(qhit,'\0');
            qhit=subBuffer(qhit,endIndex+1,qhit.length);
            num++;
            if (qhit==null)
                break;
        }

        cResult.addNumOfFiles(numOfFiles);

    }

    public int bufferIndexOf(byte[] buffer,char c){

        for (int i=0;i<buffer.length;i++){
            if (buffer[i]==c){
                return i;
            }
        }
        return -1;
    }

}
